package com.example.springbootkafka.kafka;


import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.streams.processor.PunctuationType;
import org.apache.kafka.streams.processor.api.Processor;
import org.apache.kafka.streams.processor.api.ProcessorContext;
import org.apache.kafka.streams.processor.api.ProcessorSupplier;
import org.apache.kafka.streams.processor.api.Record;

import java.time.Duration;

/**
 * 延时消息处理器
 *
 * @author huangchangjin
 * @date 2025/1/8 15:16
 */
@Slf4j
public class DelayProcessorSupplier implements ProcessorSupplier<String, String, String, String> {
    @Override
    public Processor<String, String, String, String> get() {
        return new Processor<String, String, String, String>() {
            private ProcessorContext<String, String> context;

            @Override
            public void init(ProcessorContext<String, String> context) {
                this.context = context;
            }


            @Override
            public void process(Record<String, String> record) {
                long delay = 5000; // 延迟 5 秒
                context.schedule(Duration.ofMillis(delay), PunctuationType.WALL_CLOCK_TIME, (timestamp) -> {
                    log.info("timestamp = {}", timestamp);
                    context.forward(record,"");
                });

            }

            @Override
            public void close() {
                // 资源清理
            }
        };
    }
}
